package com.aispeech.asr.apis.detection.function;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.Function;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import com.aispeech.asr.apis.detection.config.AppConfig;
import com.aispeech.asr.apis.detection.config.DopcConfig;
import com.aispeech.asr.apis.detection.domain.Audio;
import com.aispeech.asr.apis.detection.domain.EventDetection;
import com.aispeech.asr.apis.detection.domain.EventDetectionContext;
import com.aispeech.asr.apis.detection.domain.EventDetectionResult;
import com.aispeech.asr.apis.detection.exceptions.BusinessException;
import com.aispeech.asr.apis.detection.form.DetectionForm;
import com.aispeech.asr.apis.detection.vo.SedVo;
import com.aispeech.asr.clib.arpc.EARpcState;
import com.aispeech.asr.clib.arpc.IARpcCallback;
import com.aispeech.asr.clib.decoder.EAudioCodec;
import com.aispeech.asr.comm.enums.ErrorCode;
import com.aispeech.jarpc.ArpcException;
import com.aispeech.jarpc.JArpc;
import com.aispeech.jarpc.message.ArpcMessage;
import com.aispeech.jarpc.op.sed.SedRequest;
import com.alibaba.fastjson.JSONArray;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.type.CollectionType;
import com.sun.jna.Pointer;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;


@Slf4j
@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class EventDetectionFunction implements Function<EventDetectionContext, EventDetectionContext>, IARpcCallback {
	
	private final static int BUFF_LEN = 3200;
	
	@Autowired
    AppConfig config;
	@Autowired
    ObjectMapper objectMapper;
	@Autowired
	DopcConfig DOpsConfig;
	
	
	private BusinessException exception;
	private EventDetectionContext context;

    //private ARpc rpc;
    private JArpc rpc;
    private String arpcSessionId;
    private String sessionId;
	
	public void callback(String arpcSessionId, int state, Pointer data, int size, Pointer userData) {
    	log.info("[ EventDetection.onFeed ] session: {}, arpcSessionId: {},state: {}",sessionId,arpcSessionId,state);
		
		if(state == EARpcState.S_OK.getValue() || state == EARpcState.S_EOF.getValue()) {
			if (size > 0) {
            	log.info("[ EventDetection.callback ] session: {}, arpcSessionId: {},size: {}",sessionId,arpcSessionId,size);
            	byte[] bytes = data == null ? new byte[0] : data.getByteArray(0, size);
            	
            	try {
	                JsonNode res = objectMapper.readTree(bytes, 0, size);
	                int code = res.get("code").asInt();
	                
	                log.info("[ EventDetection.callback feed callback info ] session: {}, arpcSessionId: {},code: {}",sessionId,arpcSessionId,code);

	                if (code == 0) {
	                	EventDetectionResult result = objectMapper.readValue(bytes, 0, size, EventDetectionResult.class);
	                	context.setResult(result);
	                } else {
	                    exception = new BusinessException(String.format("EventDetection failed. session: %s,arpcSessionId: %s,code: %s, bytes: %s", sessionId,arpcSessionId,code, bytes));
	                }
	            } catch (Exception e) {
	                exception = new BusinessException(String.format("parse EventDetection result failed. session: %s,arpcSessionId: %s, bytes: %s, error: %s",
	                		sessionId,arpcSessionId, bytes, e.getMessage()));
	            }
            } else {
                exception = new BusinessException(String.format("[ EventDetection.callback ] session: %s,arpcSessionId: %s, state: %d, size length is zero", sessionId, arpcSessionId,state));
            }
		} else {
			byte[] bytes = data == null ? new byte[0] : data.getByteArray(0, size);
			exception = new BusinessException(String.format("[ EventDetection.callback ] session: %s,arpcSessionId: %s, state: %d, reason: %s",
					sessionId, arpcSessionId, state, new String(bytes, StandardCharsets.UTF_8)));
		}
	}
	
	
    @SneakyThrows
    @Override
    public EventDetectionContext apply(EventDetectionContext context) {
    	
    	context.addMetric("detection_begin");
    	
    	this.context = context;
    	this.exception = null;
    	this.sessionId = context.getSessionId();

        DetectionForm form = context.getForm();
    	
    	Audio audio= form.getAudio();
    	
    	
    	String audioType = audio.getAudioType();
//    	Integer sampleRate = audio.getSampleRate() == 8000 ? 8000 : 16000;
//    	Integer channel = audio.getChannel();
//    	Integer sampleBytes = audio.getSampleBytes();
    	EAudioCodec codec = StringUtils.isBlank(audioType) ? EAudioCodec.from("wav") : EAudioCodec.from(audioType);
    	
    	if (codec == null) {
            throw new BusinessException(ErrorCode.ERR_AUDIO_SPLIT_FAILED.errid, "un-supported audio type: " + form.getAudio().getAudioType());
        }
    	
    	try {
    		doProcess(codec,form);
    		
//    		prepareAndStartEventDetection();
//          doEventDetection(codec);
    	}catch(Exception e) {
    		throw e;
    	}finally {
    		rpc.close();
    		rpc = null;
    		log.info("RPC close sessionid : {}", sessionId);
		}
    	
        context.addMetric("detection_end");

        return context;
    }
    
    
    private void doProcess(EAudioCodec codec,DetectionForm form) throws IOException, ArpcException {
    	rpc = new JArpc(DOpsConfig.getHost(),DOpsConfig.getPort(),3*60);
    	
		int skipLength = getWavHeaderLen(context.getStream());
    	
        InputStream audioStream = context.getStream();
    	
        if (codec == EAudioCodec.WAV) {
        	audioStream.skip(skipLength);
        }
        log.info("[ doProcess ]  sessionid: {}, feed length: {}...", sessionId, audioStream.available());
        
        String cfgRes = "";
        if(StringUtils.isBlank(form.getModeType())) {
        	cfgRes = config.getCfgRelationshipMap().get("default");
        }else {
        	cfgRes = config.getCfgRelationshipMap().get(form.getModeType());
            if(StringUtils.isBlank(cfgRes)) {
            	cfgRes = config.getCfgRelationshipMap().get("default");
            }
        }
        log.info("[ doProcess ]  sessionid: {}, cfgRes: {}", sessionId, config.getResRoot()+cfgRes);
        
    	SedRequest sedRequest = new SedRequest(audioStream,config.getResRoot()+cfgRes);
        ArpcMessage message = rpc.process(sessionId, sedRequest);
        
//        CollectionType listType = objectMapper.getTypeFactory().constructCollectionType(ArrayList.class, EventDetection.class);
//        List<EventDetection> resultList = objectMapper.readValue(message.getBody(), listType);
        
        List<EventDetection> resultList = JSONArray.parseArray(new String(message.getBody(), "utf-8"), EventDetection.class);
        
        context.setResultList(resultList);
        
        //audioStream.close();
        
//        EventDetectionResult result = objectMapper.readValue(message.getBody(), EventDetectionResult.class);
//        context.setResult(result);
    }
    
//    @SneakyThrows
//    private void prepareAndStartEventDetection()  {
//        try {
//        	String params = "{\"op\": \"sed\"}";
//            
//            this.rpc = new ARpc();
//        	this.rpc.start(config.getAiopsHost(), config.getAiopsPort(), this, params, 3,sessionId);
//            arpcSessionId = rpc.sessionId();
//            log.info("[ prepareAndStartEventDetection ] sessionid: {},arpcSessionId: {}, params: {}", sessionId, arpcSessionId, params);
//        }catch(Exception e) {
//        	//BusinessException exception = new BusinessException(String.format("[ EventDetection.prepareAndStartEventDetection ] session: %s,arpcSessionId: %s, error message:%s", sessionId,arpcSessionId,e.getMessage()));
//        	log.error("[ prepareAndStartEventDetection ] sessionid: {},arpcSessionId: {}, error message: {}", sessionId, arpcSessionId, e.getMessage());
//        	throw e;
//        }
//    }
//    
//    @SneakyThrows
//    private void doEventDetection(EAudioCodec codec) {
//    	
//    	try {
//    		int timeout = 10;
//    		
//    		
//    		int skipLength = getWavHeaderLen(context.getAudioFile());
//    		
//        	InputStream audioStream = new FileInputStream(context.getAudioFile());
//        	
//            byte[] buffer = new byte[40*BUFF_LEN];
//            if (codec == EAudioCodec.WAV) {
//            	audioStream.skip(skipLength);
//            }
//
//            int len;
//            while ((len = audioStream.read(buffer)) > 0) {
//                rpc.feed(buffer, len);
//                log.info("[ doEventDetection ]  sessionid: {},arpcSessionId: {} feed length: {}...", sessionId,arpcSessionId, len);
//                if (exception != null)
//                    throw exception;
//            }
//            
//            audioStream.close();
//            
//            ArpcOutput arpcOutput = rpc.stop();
//            
//            log.info("[ doEventDetection ] waiting... sessionid: {},arpcSessionId: {}, code:{}", sessionId, arpcSessionId, arpcOutput.getCode());
//            
//            if(arpcOutput != null) {
//            	if(arpcOutput.getCode() == EARpcState.S_EOF.getValue()) {
//            		
//                	EventDetectionResult result = objectMapper.readValue(arpcOutput.getData(), EventDetectionResult.class);
//                	context.setResult(result);
//                    
//            	}else {
//            		BusinessException exception = new BusinessException(String.format("[ EventDetection.doEventDetection ] session: %s,arpcSessionId: %s, state:%s", sessionId,arpcSessionId,arpcOutput.getCode()));
//                	throw exception;
//            	}
//            }
//            
//            log.info("[ doEventDetection ] finished sessionid: {}, arpcSessionId: {}", sessionId,arpcSessionId);
//    	}catch(BusinessException e) {
//    		throw e;
//    	}catch(Exception e) {
//        	//BusinessException exception = new BusinessException(String.format("[ EventDetection.prepareAndStartEventDetection ] session: %s,arpcSessionId: %s, error message:%s", sessionId,arpcSessionId,e.getMessage()));
//    		log.error("[ doEventDetection ] sessionid: {}, arpcSessionId: {},timeout: {}", sessionId,arpcSessionId, e.getMessage());
//    		throw e;
//        }
//    	
//    }
    
    private int getWavHeaderLen(InputStream stream) {
    	boolean findWavEnd = false;
        int wavHeadLen = 44;
        
        InputStream audioStream = null;
        try {
        	byte[] b= org.apache.commons.io.IOUtils.toByteArray(stream);
        	audioStream = new ByteArrayInputStream(b);
        	byte[] readByte = new byte[4];
        	//char[] wavEndChar = {0x64, 0x61, 0x74, 0x61}; //data
        	byte[] wavEndByte = {100, 97, 116, 97}; //data
        	//int offset = num-8;
        	int offset = 36;
        	
        	while(!findWavEnd) {
        		audioStream.skip(offset);
				audioStream.read(readByte, 0, 4);
				//System.out.println(Arrays.toString(readByte));
        		if(Arrays.equals(readByte,wavEndByte)) {
        			findWavEnd = true;
        			break;
        		}
        		offset++;
        		wavHeadLen++;
        		audioStream.reset();
        	}
        	if (stream.markSupported())
            	stream.reset();
        }catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			if (null != audioStream) {
				try {
					audioStream.close();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					log.error("[getWavHeaderLen] get audio header failed. sessionid: {}, arpcSessionId: {}",sessionId,arpcSessionId);
					e.printStackTrace();
					
				}
			}
		}
        
    	return wavHeadLen;
    	
    }

}
